package com.gzhryc.shared_device.oem.micro.iot;

import com.gzhryc.common.json.JsonTools;
import com.gzhryc.common.kafka.JFKafkaProducer;
import com.gzhryc.common.kafka.models.KafkaMessage;
import com.gzhryc.common.logging.Logger;
import com.gzhryc.shared_device.iot.code.IIOTLogServer;
import com.gzhryc.shared_device.iot.code.dao.db.IOTLogsRecord;
import com.gzhryc.shared_device.iot.code.services.IOTLogsRecordService;
import com.gzhryc.shared_device.cdcz.code.dao.db.CDCZSlotInstantLog;
import com.gzhryc.shared_device.cdcz.code.services.CDCZSlotInstantLogService;
import com.gzhryc.shared_device.iot.code.IIOTLogEntity;

import java.util.concurrent.*;

/**
 * IOT日志服务
 * @version 2023-01-13
 */
public class IOTLogServer implements IIOTLogServer {
	
	static Logger log = Logger.getLogger(IOTLogServer.class);

	static IOTLogServer me;

	public static IOTLogServer self(){
		if(me == null){
			me = new IOTLogServer();
		}
		return me;
	}

	ThreadPoolExecutor executorService;		//工作线程池
	JFKafkaProducer producer;

	private IOTLogServer(){}

	public void start(){
		if(executorService != null) {
			executorService.shutdown();
			executorService = null;
		}
		int poolSize = Runtime.getRuntime().availableProcessors();
		BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(1024);
		RejectedExecutionHandler policy = new ThreadPoolExecutor.AbortPolicy();	//当新任务被线程池拒绝时，会抛出RejectedExecutionException
		//可堵塞的线程池队列
		executorService = new ThreadPoolExecutor(poolSize, poolSize * 2, 180, TimeUnit.SECONDS,queue,policy);
	}

	public void send(IIOTLogEntity logEntity) {
		if(logEntity != null) {
			if(executorService != null) {
				Runnable workThread = new NormalWorkThread(logEntity,producer);
				try {
					executorService.execute(workThread);
				} catch (Exception e) {
					log.error("执行日志线程失败，丢失日志：{{1:json}}", e, logEntity);
				}
			}else{
				log.error("IOT日志服务未初始化");
			}
		}
	}

	static class NormalWorkThread implements Runnable {

		JFKafkaProducer producer;
		IIOTLogEntity entity;

		public NormalWorkThread(IIOTLogEntity entity, JFKafkaProducer producer){
			this.entity = entity;
			this.producer = producer;

		}

		@Override
		public void run() {
			if (producer != null) {
				try {
					KafkaMessage msg = new KafkaMessage();
					msg.setTopic("iot_" + entity.getIotId() + "_device_log");
					msg.setKey(entity.getIotId().toString());
					msg.setValue(JsonTools.toJson(entity));
					producer.send(msg);
				} catch (Exception e) {
					log.error("发送IOT日志错误", e);
				}
			}else if(IOTLogsRecord.class.isInstance(entity)){
				IOTLogsRecordService service = IOTLogsRecordService.self();
				if (service != null) {
					IOTLogsRecord record = IOTLogsRecord.class.cast(entity);
					if (!service.add(record)) {
						log.error("日志入库失败，丢失日志：{{0:json}}", entity);
					}
				} else {
					log.error("日志入库失败，未初始化数据库，丢失日志：{{0:json}}", entity);
				}
			}else if(CDCZSlotInstantLog.class.isInstance(entity)){
				CDCZSlotInstantLogService service = CDCZSlotInstantLogService.self();
				if (service != null) {
					CDCZSlotInstantLog instantLog = CDCZSlotInstantLog.class.cast(entity);
					if (!service.add(instantLog)) {
						log.error("日志入库失败，丢失日志：{{0:json}}", instantLog);
					}
				} else {
					log.error("日志入库失败，未初始化数据库，丢失日志：{{0:json}}", entity);
				}
			}
		}
	}
}
